package com.fstec.ks.master.queue;

import io.netty.handler.codec.DecoderException;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.sinotn.SinotnLogger;

public class LocalFileWriter {
	private static Logger LOG=LoggerFactory.getLogger("com.sinotn.ks");
	public static final String META_FILE="write.m";
	// 头文件20字节，
	//8字节[0-7]记录写入位置;
	//8字节[8-15]记录总记录数;
	//4字节[16-19]记录单数据文件大小;
	public static final int HEAD_SIZE=20;
	private boolean trace=LOG.isTraceEnabled();
	private RandomAccessFile headRaf;
	private FileChannel headChannel;
	private RandomAccessFile dataRaf;
	private FileChannel dataChannel;
	private MappedByteBuffer headMappedByteBuffer;
	// 总写入偏移位置
	private AtomicLong writePos;
	// 当前文件写入偏移位置
	private AtomicLong curWritePos=new AtomicLong(0);
	// 总写入记录数
	private AtomicLong receiveCount;
	private boolean genNewDataFile=false;
	private File folder;
	private int fileSize;
	private String dtname;
	private ByteBuffer rcvBuffer;
	private long mark=-1;
	private String id;
	
	public LocalFileWriter(File file, int fileSize, int rcvBufferSize){
		this.folder=file;
		this.fileSize=fileSize;
		this.rcvBuffer=ByteBuffer.allocate(rcvBufferSize);
		this.id=file.getName();
	}
	
	private void initDataFile(long offset) throws IOException{
		if(this.dataChannel!=null){
			this.dataChannel.force(false);
			this.dataChannel.close();
			this.dataRaf.close();
		}
		long t=System.currentTimeMillis();
		File dtfile=new File(this.folder, ReceiverUtil.formatOffset(offset));

		this.dataRaf=new RandomAccessFile(dtfile, "rw");
		this.dataChannel=this.dataRaf.getChannel();
		final int bufSize=1024*1024;// 初始化缓存1M
		ByteBuffer buffer=ByteBuffer.allocate(bufSize);
		for(long i=this.fileSize/bufSize;i>0;i--){
			this.dataChannel.write(buffer);
			buffer.rewind();
		}
		this.dataChannel.force(false);
		this.dataChannel.position(0);
		// 总写入偏移位置
		this.writePos.set(offset);
		// 当前文件偏移位置
		this.curWritePos.set(0);
		this.dtname=dtfile.getName();
		LOG.info("初始新数据文件{},writePos={},curWritePos={},receiveCount={}, {}ms",(this.folder.getName()+"/"+dtfile.getName()),this.writePos.get(),this.curWritePos,this.receiveCount.get(),(System.currentTimeMillis()-t));
	}
	/**
	 * 获取写入记录数
	 * @return
	 * @auth <a href="mailto:xieqj@fstec.com">谢启进</a>
	 * @date 2023年10月18日 下午2:59:25
	 */
	public long getWriteCount(){
		if(this.receiveCount==null){
			return 0;
		}
		return this.receiveCount.get();
	}
	
	/**
	 * 
	 * @return 1 新目录初始化， 0旧目录初始化
	 * @auth <a href="mailto:xieqj@fstec.com">谢启进</a>
	 * @date 2022年5月14日 下午5:16:02
	 */
	public int init(){
		// 写消息初始化
		File head=new File(this.folder, META_FILE);
		boolean ok=false;
		try{
			boolean writeMetaExist=head.exists();
			this.headRaf=new RandomAccessFile(head, "rw");
			this.headChannel=this.headRaf.getChannel();
			this.headMappedByteBuffer=this.headChannel.map(MapMode.READ_WRITE, 0, HEAD_SIZE);
			int iRet=0;
			if(writeMetaExist){// 已存在头文件
				// 总写入偏移位置
				this.writePos=new AtomicLong(this.headMappedByteBuffer.getLong(0));
				// 总写入记录数
				this.receiveCount=new AtomicLong(this.headMappedByteBuffer.getLong(8));
				// 单文件
				int oldFileSize=this.headMappedByteBuffer.getInt(16);
				if(oldFileSize!=this.fileSize){
					LOG.info("[queue={}]数据文件大小从write.m读取{}覆盖系统配置{}",this.folder.getName(), oldFileSize,this.fileSize);
					this.fileSize=oldFileSize;
				}
				LOG.info("[queue={}]从{}读取writePos={},receiveCount={}",this.folder.getName(), META_FILE,this.writePos.get(),this.receiveCount.get());
			}else{
				this.writePos=new AtomicLong(0);
				this.receiveCount=new AtomicLong(0);
				this.headMappedByteBuffer.putInt(16, this.fileSize);
				iRet=1;
			}

			// 计算文件名偏移
			long offset=(this.writePos.get()/this.fileSize)*this.fileSize;
			File dtfile=new File(this.folder, ReceiverUtil.formatOffset(offset));
			if(dtfile.exists()){
				this.dataRaf=new RandomAccessFile(dtfile, "rw");
				this.dataChannel=this.dataRaf.getChannel();
				// 当前文件写入偏移位置
				this.curWritePos=new AtomicLong(this.writePos.get()%this.fileSize);
				this.dataChannel.position(this.curWritePos.get());
				this.dtname=dtfile.getName();
				SinotnLogger.DEBUG.info("[queue={}]挂载数据文件{},writePos={},curWritePos={},receiveCount={}",this.id ,dtfile.getName(),this.writePos.get(),this.curWritePos,this.receiveCount.get());
			}else{
				this.initDataFile(offset);
			}
			ok=true;
			return iRet;
		}catch(Throwable e){
			throw new RuntimeException(e);
		}finally{
			if(!ok){
				if(this.headChannel!=null){
					try{
						this.headChannel.close();
						this.headRaf.close();
					}catch(Throwable e){}
				}
				if(this.dataChannel!=null){
					try{
						this.dataChannel.close();
						this.dataRaf.close();
					}catch(Throwable e){}
				}
			}
		}
	}
	
	private void innerWrite(ByteBuffer buffer) throws IOException{
		// 当前数据文件可写字节数
		int maxSize=(int)(this.fileSize-this.dataChannel.position());
		// 需要写的字节数
		int dataSize=buffer.remaining();

		if(dataSize>maxSize){
			// 当前数据文件写不下当前数据
			int srcLimit=buffer.limit();
			int leftLimit=buffer.position()+maxSize;
			buffer.limit(leftLimit);
			int writeSize=this.dataChannel.write(buffer);
			SinotnLogger.DEBUG.debug("[queue={}]数据文件切分写[dataSize={},f1Size={},f1PlanWrite={},f1Write={}]",this.folder.getName(),dataSize,maxSize,leftLimit,writeSize);
			buffer.limit(srcLimit);
			// 生成新文件
			long newFilePos=((this.writePos.get()/this.fileSize)+1)*this.fileSize;
			this.initDataFile(newFilePos);
			this.genNewDataFile=true;
		}
		if(this.trace){
			SinotnLogger.DEBUG.trace("[queue={}]数据文件{}写[dataSize={}]",this.id,this.dtname,buffer.remaining());
		}

		int byteSize=this.dataChannel.write(buffer);
		// 当前数据文件写位置，注意：initDataFile有修改本属性
		this.curWritePos.addAndGet(byteSize);
		// 总数据文件写位置，注意：initDataFile有修改本属性
		this.writePos.addAndGet(byteSize);
	}
	
	public void flush(){
		// 记录执行数据写盘
		try{
			// 数据文件写盘
			this.dataChannel.force(false);
			// 队列头文件，内存映射共享
			// 总写偏移位置
			this.headMappedByteBuffer.putLong(0, this.writePos.get());
			// 总记录数
			this.headMappedByteBuffer.putLong(8,this.receiveCount.incrementAndGet());
			if(this.genNewDataFile){
				this.genNewDataFile=false;
				this.headMappedByteBuffer.force();
			}
			if(trace){
				SinotnLogger.DEBUG.trace("消息队列写盘[file={}/{}, writePos={}, RcvCount={}]",this.id,this.dtname,this.writePos.get(),this.receiveCount.get());
			}
			this.mark=-1;
		}catch(Throwable e){
			throw new RuntimeException("执行队列数据写盘失败"+this.id+"/"+this.dtname, e);
		}
	}
	
	public void write(byte[] data) throws IOException {
		int len=data.length;
		int transferSize;
		int offset=0;
		this.rcvBuffer.clear();
		while(len>0){
			transferSize=this.rcvBuffer.remaining();
			if(transferSize>len){
				transferSize=len;
			}
			this.rcvBuffer.put(data, offset, transferSize);
			offset+=transferSize;
			len=len-transferSize;
			this.rcvBuffer.flip();
			this.innerWrite(this.rcvBuffer);
			this.rcvBuffer.clear();
		}
	}

	public void write(byte byteVal) throws IOException {
		this.rcvBuffer.clear();
		this.rcvBuffer.put(byteVal);
		this.rcvBuffer.flip();
		this.innerWrite(this.rcvBuffer);
	}
	public void write(int intVal) throws IOException {
		this.rcvBuffer.clear();
		this.rcvBuffer.putInt(intVal);
		this.rcvBuffer.flip();
		this.innerWrite(this.rcvBuffer);
	}

	public void write(long longVal) throws IOException {
		this.rcvBuffer.clear();
		this.rcvBuffer.putLong(longVal);
		this.rcvBuffer.flip();
		this.innerWrite(this.rcvBuffer);
	}

	public void mark() {
		this.mark=this.writePos.get();
	}

	public long rollback() {
		if(this.mark==-1) return 0;
		long offset=(this.mark/this.fileSize)*this.fileSize;
		String filename=ReceiverUtil.formatOffset(offset);
		try{
			long iRet=this.writePos.get()-this.mark;
			if(!filename.equals(this.dtname)){
				// 退回到前一个数据文件
				this.dataChannel.close();
				this.dataRaf.close();
				File dtfile=new File(this.folder, this.dtname);
				boolean b=dtfile.delete();
				SinotnLogger.DEBUG.info("[queue={}]退回到前一个数据文件{}，删除{}[state={}]", this.id, filename, this.dtname, b);
				this.dtname=filename;

				dtfile=new File(this.folder, this.dtname);
				this.dataRaf=new RandomAccessFile(dtfile, "rw");
				this.dataChannel=this.dataRaf.getChannel();
			}
			
			SinotnLogger.DEBUG.info("消息回滚[queue={}, dtname={}, mark={}，pos={}]", this.folder.getName(), this.dtname, this.mark, this.writePos.get());
			
			long rollbackCurPos=(this.mark%this.fileSize);
			this.dataChannel.position(rollbackCurPos);
			this.writePos.set(this.mark);
			this.curWritePos.set(rollbackCurPos);
			this.mark=-1;
			return iRet;
		}catch(Throwable e){
			throw new DecoderException("[queue="+this.id+"]消息回滚mark="+this.mark+"，pos="+this.writePos.get()+"失败", e);
		}
	}

	public void close() {
		if(this.rcvBuffer!=null){
			this.rcvBuffer.clear();
			this.rcvBuffer=null;
		}
		if(this.headMappedByteBuffer!=null){
			try{
				ByteBufferUtil.clean(this.headMappedByteBuffer);
			}catch(Throwable e){}
			this.headMappedByteBuffer=null;
		}
		if(this.headChannel!=null){
			try{
				this.headChannel.close();
			}catch(Throwable e){}
			this.headChannel=null;
		}
		if(this.headRaf!=null){
			try{
				this.headRaf.close();
			}catch(Throwable e){}
			this.headRaf=null;
		}
		if(this.dataChannel!=null){
			try{
				this.dataChannel.close();
			}catch(Throwable e){}
			this.dataChannel=null;
		}
		if(this.dataRaf!=null){
			try{
				this.dataRaf.close();
			}catch(Throwable e){}
			this.dataRaf=null;
		}
	}
}
